package com.demo.webflux.jpaadapter;

import org.reactivestreams.Publisher;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.util.function.Tuple2;

public class RxBookRepository extends ReactiveCrudRepositoryAdapter<Book, Integer, BookJpaRepository> {

	public RxBookRepository(BookJpaRepository delegate, Scheduler scheduler) {
		super(delegate, scheduler);
	}

	// 接收两个响应式流并使用zip操作订阅他们，当两个流的值都准备就绪时，在delegate实例上调用相应的方法，并将阻塞执行转移到专用调度程序。
	public Flux<Book> findByIdBetween(Publisher<Integer> lowerPublisher, Publisher<Integer> upperPublisher) {
		return Mono.zip(Mono.from(lowerPublisher), Mono.from(upperPublisher))
				.flatMapMany((Tuple2<Integer, Integer> data) -> {
					int low = data.getT1();
					int upp = data.getT2();
					return Flux.fromIterable(delegate.findByIdBetween(low, upp))
							.subscribeOn(scheduler);
				})
				.subscribeOn(scheduler);
	}

	// 在专用调度程序上调用相应的方法，并将Iterable映射到Flux
	public Flux<Book> findShortestTitle() {
		return Mono.fromCallable(delegate::findShortestTitle)
				.subscribeOn(scheduler)
				.flatMapMany(Flux::fromIterable);
	}

}
